package com.isesol.ime.service.mqtt;

import com.isesol.arch.common.utils.*;
import com.isesol.ime.*;
import com.isesol.ime.service.*;
import com.isesol.ime.util.*;
import lombok.extern.slf4j.*;
import org.fusesource.hawtbuf.*;
import org.fusesource.mqtt.client.*;
import org.springframework.stereotype.*;

import javax.annotation.*;

@Component
@Slf4j
public class MqttService {

    private String mqttIP;

    private String mqttPort;

    private String mqttUserName;

    private String mqttPwd;

    private CallbackConnection mqttConnection;

	@Resource
	FeedBackService feedBackService;

    @Resource
    SubFeedbackService subFeedbackService;

    public void publish(final String topic, String msg){

        final Buffer mqttMsg = new UTF8Buffer(msg);

        final UTF8Buffer mqttTopic = new UTF8Buffer(topic);

        mqttConnection.publish(mqttTopic, mqttMsg, QoS.AT_LEAST_ONCE, false, new Callback<Void>(){

            public void onSuccess(Void v){

                log.info("publish {} success\n{}", topic,mqttMsg.toString());
            }

            public void onFailure(Throwable value){


	            log.error("mqtt publish fail",value);
            }
        });

    }

    @PostConstruct
    public void init() throws Exception {

        if(PropertyFileUtil.get("system.env").equals("local")){
            return ;
        }
        if (EnvUtil.sendWithMqtt()) {

            mqttIP = PropertyFileUtil.get(Constant.MQTT_IP);
            mqttPort = PropertyFileUtil.get(Constant.MQTT_PORT);
            mqttUserName = PropertyFileUtil.get(Constant.MQTT_USERNAME);
            mqttPwd = PropertyFileUtil.get(Constant.MQTT_PASS);

            MQTT mqtt = new MQTT();
            mqtt.setHost(mqttIP, Integer.valueOf(mqttPort));
            mqtt.setUserName(mqttUserName);
            mqtt.setPassword(mqttPwd);
            mqtt.setKeepAlive((short) 30);
            mqtt.setConnectAttemptsMax(-1);
            mqtt.setReconnectBackOffMultiplier(1);
            mqtt.setReconnectDelay(100);
            // mqtt.setCleanSession(false);
            mqtt.setClientId("imeconfig-" + System.currentTimeMillis());


            final CallbackConnection connection = mqtt.callbackConnection();

            connection.listener(new Listener() {

                public void onConnected() {

                    log.info("mqtt connected");
                    mqttConnection = connection;

                }

                public void onDisconnected() {

                    log.warn("mqtt disConnected");
                }

                public void onFailure(Throwable value) {

                    log.error("mqtt error", value);
                }

                public void onPublish(UTF8Buffer topic, Buffer msg, Runnable ack){

                    new Thread(ack).start();

	                log.info("receive mqtt msg,{}\n{}",topic.utf8().toString(),msg.utf8().toString());

                    if(topic.utf8().toString().equals(Constant.MQTT_MACHINE_ONLINE_TOPIC)){
                        try{
                            subFeedbackService.compensate(msg.utf8().toString());
                         } catch (Exception e){
                            log.error("mqtt msg process error",e);
                        }
                    }else	if(topic.utf8().toString().equals(Constant.MQTT_FEEDBACK)){
						try{
							feedBackService.consume(msg.utf8().toString());
						} catch (Exception e){
							log.error("mqtt msg process error",e);
						}
					}

                }
            });

            final Topic[] topics = {
                    new Topic(Constant.MQTT_MACHINE_ONLINE_TOPIC, QoS.AT_LEAST_ONCE),
                    new Topic(Constant.MQTT_FEEDBACK, QoS.AT_LEAST_ONCE)
            };

            connection.connect(new Callback<Void>(){

                @Override
                public void onSuccess(Void value){

                    connection.subscribe(topics, new Callback<byte[]>(){

                        public void onSuccess(byte[] qoses){

                            log.info("mqtt subscribe success");

                        }

                        public void onFailure(Throwable value){

                            log.error("mqtt subscribe fail", value);
                        }
                    });
                }

                @Override
                public void onFailure(Throwable value){

                    log.error("mqtt connect fail", value);
                }
            });
        }
    }
}
